package com.sinozo.data.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.sinozo.data.bean.key.UserUniqueActionKeyByBean;
import com.sinozo.data.common.ConfigConstant;
import com.sinozo.data.common.DataConfig;
import com.sinozo.data.utils.DateFormatUtil;
import com.sinozo.data.utils.MyKafkaUtil;
import com.sinozo.data.utils.PathUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author zhaoyb
 * @version 1.0
 * @description 独立用户动作日志->页面级别event
 * @date 2023/5/16 19:42:34
 */
public class DwdTrafficUniqueEventActionDetail {

    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(DwdTrafficUniqueEventActionDetail.class);
        try {
            //读取环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(DataConfig.KAFKA_PARTITION_NUM);

            //开启checkpoint
            env.enableCheckpointing(DataConfig.FLINK_CHECKPOINT_INTERVAL, CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(DataConfig.FLINK_CHECKPOINT_TIMEOUT);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);

            //1.2 设置状态后端
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
            );
            env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(3L, TimeUnit.DAYS), Time.of(1L, TimeUnit.MINUTES)));

            String hdfsPath = DataConfig.HDFS_SERVER;
            env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(PathUtil.getHdfsPath(hdfsPath,"dwdTrafficUniqueEventActionDetail")));
            env.setStateBackend(new EmbeddedRocksDBStateBackend());

            //2.读取Kafka 启动日志主题创建流
            String topic = ConfigConstant.KAFKA_ACTION_LOG_TOPIC;
            String groupId = ConfigConstant.KAFKA_UNIQUE_EVENT_ACTION_DETAIL_GROUP_ID;

            DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

            //3.将每行数据转换为JSON对象
            SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
                @Override
                public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                    try {
                        JSONObject jsonObject = JSON.parseObject(value);
                        out.collect(jsonObject);
                    } catch (Exception e) {
                        logger.error("Error parsing", e);
                    }
                }
            });

            //4.使用状态编程实现按照device_id以及action_type和subEvent的去重
            SingleOutputStreamOperator<JSONObject> uvDS = jsonObjDS.keyBy(new KeySelector<JSONObject, UserUniqueActionKeyByBean>() {
                @Override
                public UserUniqueActionKeyByBean getKey(JSONObject value) throws Exception {
                    UserUniqueActionKeyByBean keyByBean = new UserUniqueActionKeyByBean();
                    String deviceId = value.getJSONObject("common").getString("device_id");
                    String code = value.getJSONObject("common").getString("code");
                    String actionType = value.getJSONObject("action").getString("action_type");
                    String subEvent = value.getJSONObject("action").getString("event");
                    keyByBean.setDeviceId(deviceId);
                    keyByBean.setActionType(actionType);
                    keyByBean.setSubEvent(subEvent);
                    keyByBean.setCode(code);
                    return keyByBean;
                }
            }).filter(new RichFilterFunction<JSONObject>() {
                private ValueState<String> lastActiveState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("last-event-active", String.class);
                    //设置状态的TTL
                    StateTtlConfig ttlConfig = new StateTtlConfig.Builder((Time.days(1)))
                            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                            .build();
                    stateDescriptor.enableTimeToLive(ttlConfig);
                    lastActiveState = getRuntimeContext().getState(stateDescriptor);
                }

                @Override
                public boolean filter(JSONObject value) throws Exception {
                    //获取状态数据&当前数据中的时间戳并转换为日期
                    String lastDate = lastActiveState.value();
                    JSONObject action = value.getJSONObject("action");
                    Long ts = action.getLong("ts");
                    String curDate = DateFormatUtil.toDate(ts);
                    if (lastDate == null || !lastDate.equals(curDate)) {
                        lastActiveState.update(curDate);
                        return true;
                    } else {
                        return false;
                    }
                }
            });

            //5.将数据写到Kafka
            String targetTopic = ConfigConstant.KAFKA_UNIQUE_EVENT_ACTION_TOPIC;

            uvDS.map(JSONAware::toJSONString).addSink(MyKafkaUtil.getFlinkKafkaProducer(targetTopic));

            //6.启动任务
            env.execute("DwdTrafficUniqueEventActionDetail");
        } catch (Exception e) {
            logger.error("exception", e);
        }
    }


}
